package com.lagou.rpc.balance;

import com.lagou.rpc.netty.RpcClinetChannelHolder;
import com.lagou.util.LongByteUtil;
import org.apache.zookeeper.data.Stat;
import org.springframework.stereotype.Component;

import java.io.File;
import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @ClassName ZookeeperRpcLoadBanlance
 * @Description 基于zookeeper实现的负载均衡
 * @Author playboy
 * @Date 2021/6/8 7:20 下午
 * @Version 1.0
 **/
@Component
public class ZookeeperRpcLoadBanlance extends AbstractRpcLoadBanlance {

	//维护一个rpcClinetChannelHolder和node的对应关系
	private Map<RpcClinetChannelHolder, String> holderNodeMap = new ConcurrentHashMap<>();


	/**
	 * 项目启动，启动定时任务
	 */
	@Override
	protected void start() {
		Thread thread = new Thread(new Runnable() {
			@Override
			public void run() {
				while (true) {
					try {
						Thread.sleep(5000L);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					System.out.println("开始检查是否存在过期请求");
					for (String server : servers) {
						Stat stat = new Stat();
						try {
							client.getData()
									.storingStatIn(stat)
									.forPath(path + File.separator + server);
							Long mtime = stat.getMtime();
							if (mtime.compareTo(System.currentTimeMillis() - 5000L) < 0) {
								//更新时间超过5s了,计时清零
								System.out.println(server + "节点访问时间过期");
								client.setData()
										.forPath(path + File.separator + server, LongByteUtil.longToBytes(0L));
							}
						} catch (Exception e) {
							e.printStackTrace();
						}

					}
				}
			}
		});
		thread.start();
	}

	@Override
	public RpcClinetChannelHolder getChannel() {
		String server = "";
		Long time = Long.MAX_VALUE;
		for (String node : servers) {
			try {
				byte[] bytes = client.getData()
						.forPath(path + File.separator + node);
				Long aLong = Long.valueOf(LongByteUtil.bytesToLong(bytes));
				InetSocketAddress inetSocketAddress = maps.get(node).getInetSocketAddress();
				System.out.println("节点：" + inetSocketAddress.getHostString() + ":" + inetSocketAddress.getPort() + "上次访问耗时：" + aLong + "毫秒");
				if (aLong.compareTo(time) < 0) {
					time = aLong;
					server = node;
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		InetSocketAddress inetSocketAddress = maps.get(server).getInetSocketAddress();
		System.out.println("当前访问的地址：" + inetSocketAddress.getHostString() + ":" + inetSocketAddress.getPort());
		return maps.get(server);
	}


	@Override
	public void setResponseTime(RpcClinetChannelHolder rpcClinetChannelHolder, long time) {
		String node = holderNodeMap.get(rpcClinetChannelHolder);
		//更新节点时间到zookeeper
		try {
			client.setData()
					.forPath(path + File.separator + node, LongByteUtil.longToBytes(time));
		} catch (Exception e) {
			e.printStackTrace();
		}
	}


	@Override
	protected void addFinishHandler(String node, RpcClinetChannelHolder rpcClinetChannelHolder) {
		holderNodeMap.put(rpcClinetChannelHolder, node);
	}

	@Override
	protected void subFinishHandler(String node, RpcClinetChannelHolder rpcClinetChannelHolder) {
		holderNodeMap.remove(rpcClinetChannelHolder, node);
	}


}
